// -*-mode:c++; coding:utf-8-*-

#ifndef _XBASE_ROW_MUTATION_HPP_
#define _XBASE_ROW_MUTATION_HPP_

#include <deque>

#include <boost/thread/mutex.hpp>

#include "error_code.hpp"

#include "protocol_struct.hpp"
#include "row_mutation_data.hpp"

namespace xbase
{
	// usage:
	//    protocol::table_info_ptr ptr;
	//    async_get_table_info(ptr, "table_name");
	//    row_mutation mutation(ptr, "row_name"); // create row mutation for this table
	//    ec = mutation.add("column_i", "my_data"); // 加入数据时对列名进行验证
	//    async_apply(mutation, handler);
	//    async_apply_on_tablet(session, tablet_name, mutation, timestamp, handler);
	
	// 
	// 仅是对用户的封装，对row_mutation_data添加记录时通过该类进行
	// 检查，检查包含：
	// 
	// - 行名不能为空
	// - 列名不能包含正则表达式（仅查询时允许）
	// - 如果不是删除记录（!value.empty()），则列名不能为空、小列名
	//   不能为空（列名为空表示删除整行、小列名为空表示删除整个列族）
	// - 按照指定表结构信息，列族及所属的组存在（若是删除记录且列为空，则不检查）
	//
	// 删除一行的所有记录时，需要在所有组下面都添加一条记录
	// 
	class row_mutation
	{
	public:
		typedef boost::system::error_code error_code;
		
	public:
		row_mutation() {}
		
		row_mutation(protocol::table_info_ptr table_info,
			     const std::string &row_name)
			: m_table_info(table_info)
			, m_data(new row_mutation_data(row_name)) {}

		// data MUST be empty
		row_mutation(protocol::table_info_ptr table_info,
			     const row_mutation_data_ptr &data)
			: m_table_info(table_info)
			, m_data(data) {
			assert(data->empty());
		}

		const std::string &table_name() const {
			return m_table_info->m_table_name;
		}

		const std::string &row_name() const {
			return m_data->row_name();
		}

		bool empty() const {
			return m_data->empty();
		}

		void clear() const {
			m_data->clear();
		}

		const row_mutation_data &data() const {
			return *m_data;
		}

		row_mutation_data_ptr data_ptr() const {
			return m_data;
		}

		void reset() {
			m_table_info.reset();
			m_data.reset();
		}

		bool operator!() {
			return (! m_table_info) &&
				(! m_data);
		}

		error_code change_row_name(const std::string &row_name) const {
			error_code ec;
			if(row_name.empty())
				ec = error::bad_row_name;
			else
				m_data->change_row_name(row_name);
			return ec;
		}

		error_code add(const std::string &column_name,
			       const std::string &value,
			       uint64_t timestamp = NO_TIMESTAMP) const;

		// alias for add

		error_code insert(const std::string &column_name,
				  const std::string &value,
				  uint64_t timestamp = NO_TIMESTAMP) const {
			return add(column_name, value, timestamp);
		}

		error_code update(const std::string &column_name,
				  const std::string &value,
				  uint64_t timestamp = NO_TIMESTAMP) const {
			return add(column_name, value, timestamp);
		}
		
		error_code remove(const std::string &column_name,
				  uint64_t timestamp = NO_TIMESTAMP) const {
			return add(column_name, std::string(), timestamp);
		}

		// // 将另一个行修改对象的内容合并过来，且需要通过检查；若
		// // 没有通过检查，则自身不变。
		// error_code merge_from(const row_mutation &another) {
		// 	error_code ec;
		// 	// 表结构不同，则先检查
		// 	if((m_table_info.get() != another.m_table_info.get()) &&
		// 	   (*m_table_info != *another.m_table_info))
		// 	{
		// 		ec = another.check_with_table_info(*m_table_info);
		// 	}
		// 	if(! ec) // 没有问题，再合并
		// 	{
		// 		m_data->merge_from(another.m_data);
		// 	}
		// 	return ec;
		// }

		// error_code check_with_table_info(const table_info &schema) const {
		// 	error_code ec;
		// 	return ec;
		// }

	private:
		protocol::table_info_ptr m_table_info;
		row_mutation_data_ptr m_data;
	};

	class multi_row_mutation
	{
	public:
		typedef boost::system::error_code error_code;
		typedef std::map<std::string, row_mutation_data_ptr> data_map;

	public:
		explicit
		multi_row_mutation(protocol::table_info_ptr table_info)
			: m_table_info(table_info) {}

		error_code add(const std::string &row_name,
			       const std::string &column_name,
			       const std::string &value,
			       uint64_t timestamp = NO_TIMESTAMP) {
			error_code ec;
			if(row_name.empty())
			{
				ec = error::bad_row_name;
			}
			else
			{
				data_map::iterator it = m_map.find(row_name);
				if(it == m_map.end())
				{
					row_mutation_data_ptr ptr = get_row_mutation_data_ptr(row_name);
					row_mutation mutation(m_table_info, ptr);
					ec = mutation.add(column_name, value, timestamp);
					if(! ec)
					{
						std::pair<data_map::iterator, bool> ret =
							m_map.insert(data_map::value_type(row_name,
											  ptr));
					}
					else
					{
						m_free_queue.push_back(ptr);
					}
				}
				else
				{
					row_mutation mutation = row_mutation(m_table_info, it->second);
					ec = mutation.add(column_name, value, timestamp);
				}
			}
			return ec;
		}

		// error_code get_row_mutation(row_mutation &mutation, const std::string &row_name) {
		// 	error_code ec;
		// 	if(row_name.empty())
		// 	{
		// 		ec = error::bad_row_name;
		// 	}
		// 	else
		// 	{
		// 		row_mutation_data_ptr &ptr = m_map[row_name];
		// 		if(! ptr)
		// 		{
		// 			ptr.reset(new row_mutation_data(row_name));
		// 		}
		// 		mutation = row_mutation(m_table_info, ptr);
		// 	}
		// 	return ec;
		// }

		row_mutation_data_ptr get_row_mutation_data_ptr(const std::string &row_name) {
			row_mutation_data_ptr ptr;
			if(m_free_queue.empty())
			{
				ptr.reset(new row_mutation_data(row_name));
			}
			else
			{
				ptr = m_free_queue.front();
				m_free_queue.pop_front();
				ptr->change_row_name(row_name);
			}
			return ptr;
		}

	private:
		protocol::table_info_ptr m_table_info;
		data_map m_map;
		std::deque<row_mutation_data_ptr> m_free_queue;
	};

	class row_mutation_pool
		: private boost::noncopyable
	{
	public:
		row_mutation_pool(protocol::table_info_ptr table_info,
				 std::size_t max_size = 10000)
			: m_table_info(table_info)
			, m_max_size(max_size) {}

		void update_table_info(protocol::table_info_ptr table_info) {
			boost::mutex::scoped_lock lock(m_mutex);
			m_table_info = table_info;
		}

		xbase::row_mutation new_row_mutation(const std::string &row_name) {
			boost::mutex::scoped_lock lock(m_mutex);
			if(m_free_mutation.empty())
			{
				return row_mutation(m_table_info, row_name);
			}
			else
			{
				row_mutation_data_ptr ptr = m_free_mutation.back();
				m_free_mutation.pop_back();
				ptr->change_row_name(row_name);
				return xbase::row_mutation(m_table_info, ptr);
			}
		}

		void delete_row_mutation(xbase::row_mutation mutation) {
			boost::mutex::scoped_lock lock(m_mutex);
			if(m_free_mutation.size() < m_max_size)
			{
				mutation.clear();
				m_free_mutation.push_front(mutation.data_ptr());
			}
		}

	private:
		boost::mutex m_mutex;
		protocol::table_info_ptr m_table_info;
		std::deque<row_mutation_data_ptr> m_free_mutation;
		std::size_t m_max_size;
	};

} // namespace xbase

#endif	// _XBASE_ROW_MUTATION_HPP_
